package source.javadog.mysource;

import org.redkale.source.FilterNode;
import org.redkale.util.Attribute;
import source.javadog.util.IAttribute;

import java.io.Serializable;
import java.sql.*;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Author : manyu
 * Date: 2018/9/9 18:36
 * Description:
 */
public class MyDataJdbcSource extends MyDataSqlSource<Connection> {

    public MyDataJdbcSource(Map<String, String> dbMap) {
        super(dbMap);
    }

    @Override
    protected MyPoolSource createPoolSource(ArrayBlockingQueue<Connection> queue, Semaphore semaphore, Map<String,String> dbMap) {
        return new MyPoolJdbcSource(queue, semaphore, dbMap, this.logger);
    }

    @Override
    protected <T> CompletableFuture<T> findDB(MyEntityInfo<T> info, String sql) {
        Connection conn = null;
        try {
            conn = pool.poll();
            conn.setReadOnly(true);
            final PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            ps.setFetchSize(1);
            final ResultSet set = ps.executeQuery();
            T rs = set.next() ? info.getValue(set) : null;
            set.close();
            ps.close();
            return CompletableFuture.completedFuture(rs);
        } catch (SQLException e) {
            //todo 异常处理机制
            CompletableFuture future = new CompletableFuture();
            future.completeExceptionally(e);
            return future;
        } finally {
            if (conn != null) pool.offerConnection(conn);
        }
    }

    @Override
    protected <T> CompletableFuture<Boolean> insertDB(MyEntityInfo<T> info, T value) {
        Connection conn = null;
        try {
            conn = pool.poll();
            final String sql = info.getInsertPrepareSQL(value);
            //todo 这俩属性的意思
            conn.setReadOnly(false);
            conn.setAutoCommit(true);
            PreparedStatement prestmt = createInsertPreparedStatement(conn, sql, info, value);
            prestmt.execute();
            //数据库自动生成主键的场景
            prestmt.close();
            System.out.println(info.getType().getSimpleName() + " insert sql =" + sql);
            //todo 填充主键
            return CompletableFuture.completedFuture(Boolean.TRUE);
        } catch (SQLException e) {
            CompletableFuture future = new CompletableFuture();
            future.completeExceptionally(e);
            return future;
        } finally {
            if (conn != null) pool.offerConnection(conn);
        }
    }

    protected <T> PreparedStatement createInsertPreparedStatement(final Connection conn, final String sql, final MyEntityInfo<T> info, T value) throws SQLException {
        IAttribute<T, Serializable>[] attrs = info.insertAttributes;
        final PreparedStatement prestmt = info.autoGenerated ?
                conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS) : conn.prepareStatement(sql);

        for (int i = 0; i < attrs.length; i++) {
            IAttribute<T, Serializable> attr = attrs[i];
            Serializable val = attr.get(value);
            if (val instanceof byte[]) {
                Blob blob = conn.createBlob();
                blob.setBytes(1, (byte[]) val);
                prestmt.setObject(i+1, blob);
            } else if (val instanceof AtomicInteger) {
                prestmt.setObject(i+1, ((AtomicInteger) val).get());
            } else if (val instanceof AtomicLong) {
                prestmt.setObject(i+1, ((AtomicLong) val).get());
            } else {
                prestmt.setObject(i+1, val);
            }
        }
        //todo 批量插入，是否有单插入
        prestmt.addBatch();
        return prestmt;
    }

    @Override
    protected <T> CompletableFuture<MySheet<T>> querySheetDB(MyEntityInfo<T> info) {
        Connection conn = null;
        try {
            conn = pool.poll();
            conn.setReadOnly(true);
            final List<T> list = new ArrayList();
            final String dbtype = this.pool.getDbtype();
            if ("mysql".equals(dbtype)) {
                final String listsql = "SELECT " + "*" + " FROM " + info.getTable();
                PreparedStatement ps = conn.prepareStatement(listsql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
                ResultSet set = ps.executeQuery();
                while (set.next()) {
                    list.add(infoGetValue(info, set));
                }
                set.close();
                ps.close();
                long total = list.size();
                //todo need total
                return CompletableFuture.completedFuture(new MySheet<>(total, list));
            }
            final String sql = "SELECT " + "*" + " FROM " + info.getTable();
            conn.setReadOnly(true);
            final PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            final ResultSet set = ps.executeQuery();
            final int limit = Integer.MAX_VALUE;
            int i = 0;
            while (set.next()) {
                i++;
                list.add(info.getValue(set));
                if (limit <= i) break;
            }
            long total = list.size();
            set.close();
            ps.close();
            return CompletableFuture.completedFuture(new MySheet<>(total, list));
        } catch (SQLException e) {
            CompletableFuture future = new CompletableFuture();
            future.completeExceptionally(e);
            return future;
        } finally {
            if (conn != null) pool.offerConnection(conn);
        }

    }

    @Override
    protected <T> CompletableFuture<Boolean> deleteDB(MyEntityInfo<T> info, String sql) {
        Connection conn = null;
        try {
            conn = pool.poll();
            conn.setReadOnly(false);
            conn.setAutoCommit(true);
            //todo 参数的意义
            final PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE);
            ps.setFetchSize(1);
            //todo 返回值
            boolean execute = ps.execute();
            ps.close();
            return CompletableFuture.completedFuture(execute);
        } catch (SQLException e) {
            //todo 异常处理机制
            CompletableFuture future = new CompletableFuture();
            future.completeExceptionally(e);
            return future;
        } finally {
            if (conn != null) pool.offerConnection(conn);
        }
    }

    @Override
    protected <T> CompletableFuture<Boolean> updateDB(MyEntityInfo<T> info, T update) {
        Connection conn = null;
        try {
            conn = pool.poll();
            conn.setReadOnly(false);
            conn.setAutoCommit(true);
            final String updateSQL = info.getUpdatePrepareSQL();
            final PreparedStatement prestmt = conn.prepareStatement(updateSQL);
            final IAttribute<T, Serializable> primary = info.getPrimary();
            List<IAttribute<T, Serializable>> attrs = new ArrayList<>();
            attrs.addAll(Arrays.asList(info.updateAttributes));
            attrs.add(primary);

            for (int i = 0; i < attrs.size(); i++) {
                IAttribute<T, Serializable> attr = attrs.get(i);
                Serializable val = attr.get(update);
                if (val instanceof byte[]) {
                    Blob blob = conn.createBlob();
                    blob.setBytes(1, (byte[]) val);
                    prestmt.setObject(i+1, blob);
                } else if (val instanceof AtomicInteger) {
                    prestmt.setObject(i+1, ((AtomicInteger) val).get());
                } else if (val instanceof AtomicLong) {
                    prestmt.setObject(i+1, ((AtomicLong) val).get());
                } else {
                    prestmt.setObject(i+1, val);
                }
            }

            prestmt.addBatch();
            System.out.println(info.getType().getSimpleName() + " update sql = " + updateSQL);

            boolean execute = prestmt.execute();
            prestmt.close();
            return CompletableFuture.completedFuture(execute);
        } catch (SQLException e) {
            CompletableFuture future = new CompletableFuture();
            future.completeExceptionally(e);
            return future;
        } finally {
            if (conn != null) pool.offerConnection(conn);
        }
    }

}
